package com;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * 对kafka某个主题查看详情
 * 
 * @author kaifa002
 *
 */
@Slf4j
@Component
public class KafkaDemo {
	// 主题
	public static final String TOPIC_NAME = "proc-queue-topic";
	// 消费者组
	public static final String GROUP_ID_CONFIG = "lg-default-group";
	// 如果是集群，则用逗号分隔。
	public static final String KAFKA_BROKER_LIST = "172.31.0.75:9092";

	public static Properties getConsumeProperties(String groupId, String bootstrapServer) {
		Properties props = new Properties();
		props.put("group.id", groupId);
		props.put("bootstrap.servers", bootstrapServer);
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

		return props;
	}

	public List<Map<String, Object>> topicAndPartitionDetails(String bootstrapServer, String groupId, String topic) {
		List<Map<String, Object>> result = new ArrayList<>();
		Map<Integer, Long> endOffsetMap = new HashMap<>();
		Map<Integer, Long> commitOffsetMap = new HashMap<>();
		Properties consumeProps = getConsumeProperties(groupId, bootstrapServer);
		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumeProps);

		try {
			List<TopicPartition> topicPartitions = new ArrayList<>();
			List<PartitionInfo> partitionsFor = consumer.partitionsFor(topic);
			for (PartitionInfo partitionInfo : partitionsFor) {
				TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
				topicPartitions.add(topicPartition);
			}

			Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);
			for (TopicPartition partitionInfo : endOffsets.keySet()) {
				endOffsetMap.put(partitionInfo.partition(), endOffsets.get(partitionInfo));
			}
			for (Integer partitionId : endOffsetMap.keySet()) {
				System.out.println(String.format("at %s, topic:%s, partition:%s, logSize:%s", System.currentTimeMillis(), topic, partitionId, endOffsetMap.get(partitionId)));
			}

			// 查询消费偏移量
			for (TopicPartition topicAndPartition : topicPartitions) {
				OffsetAndMetadata committed = consumer.committed(topicAndPartition);
				commitOffsetMap.put(topicAndPartition.partition(), committed.offset());
			}

			// 累加lag
			long lagSum = 0l;
			if (endOffsetMap.size() == commitOffsetMap.size()) {
				for (Integer partition : endOffsetMap.keySet()) {
					long endOffset = endOffsetMap.get(partition);
					long commitOffset = commitOffsetMap.get(partition);
					long diffOffset = endOffset - commitOffset;
					lagSum += diffOffset;
					HashMap<String, Object> partitionMap = new HashMap<>();
					// 主题
					partitionMap.put("topic", topic);
					// 消费者组
					partitionMap.put("groupId", groupId);
					// 分区
					partitionMap.put("partition", partition);
					// 最后的偏移量
					partitionMap.put("endOffset", endOffset);
					// 提交的偏移量
					partitionMap.put("commitOffset", commitOffset);
					// 积压的消息
					partitionMap.put("diffOffset", diffOffset);
					result.add(partitionMap);
				}
			} else {
				System.out.println(topic + "主题的分区丢失。");
			}
		} finally {
			if (consumer != null) {
				consumer.close();
			}
		}
		return result;
	}

	public static void main(String[] args) {
		List<Map<String, Object>> list = new KafkaDemo().topicAndPartitionDetails(KAFKA_BROKER_LIST, GROUP_ID_CONFIG, TOPIC_NAME);
		for (Map<String, Object> map : list) {
			map.forEach((k, v) -> System.out.println(k + "=" + v));
			log.info("========================");
		}
	}

}